%%% -------------------------------------------------------------------
%%% Author  : vjache
%%% Description :
%%%
%%% Created : Dec 26, 2009
%%% -------------------------------------------------------------------
-module(beta_node_rcv_srv).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include_lib("eunit/include/eunit.hrl").
-include("../../include/erete.hrl").
-include("../../include/pipe.hrl").
-include("../../include/iter.hrl").
%% --------------------------------------------------------------------
%% API exports
%% --------------------------------------------------------------------
-export([start_link/1,
		 in_pipe/2,
		 insert/3,
		 delete/3]).

%% gen_server callbacks
-export([init/1, 
		 handle_call/3, 
		 handle_cast/2, 
		 handle_info/2, 
		 terminate/2, 
		 code_change/3]).

-record(state, 
		{
		 conf,		% a beta node server conf (see #beta_node_conf)
		 log_pipe	% a modification log pipe state
		}).

%% ====================================================================
%% API functions
%% ====================================================================

%% --------------------------------------------------------------------
%% Function: in_pipe(InputId,BetaNodeServerRef) -> pipe()
%% Description: Creates an input pipe for specified input 
%%				ID to the specified beta node server.
%% Returns: a pipe (see pipe.erl,hrl) accepting {MdfCode,Key,Value} |
%%			{MdfCode,KVList}, where: 
%%				MdfCode=ins|del, 
%%				Key=Value=term(), 
%%				KVList=[{Key,Value}]. 
%% --------------------------------------------------------------------
in_pipe(InputId,BetaNodeServerRef) ->
	pipe:simple_immutable(
	  fun()->
			  make_ref()
	  end, 
	  fun(_,{ins,K,V}) ->
			  server_call(BetaNodeServerRef, #mdf_req{mdf_code=ins,input_id=InputId,data={K,V}});
		 (_,{del,K,V})->
			  server_call(BetaNodeServerRef, #mdf_req{mdf_code=del,input_id=InputId,data={K,V}});
		 (_,{ins,KVList}) when is_list(KVList)->
			  server_call(BetaNodeServerRef, #mdf_req{mdf_code=ins,input_id=InputId,data=KVList});
		 (_,{del,KVList}) when is_list(KVList)->
			  server_call(BetaNodeServerRef, #mdf_req{mdf_code=del,input_id=InputId,data=KVList});
		 (_,BadMdf)->
			  throw({bad_mdf_req,BadMdf})
	  end, 
	  fun(_)-> ok end).
insert(InputId,BetaNodeId, KVList) ->
	ins_del(ins, InputId, BetaNodeId, KVList).

delete(InputId,BetaNodeId, KVList) ->
	ins_del(del, InputId, BetaNodeId, KVList).

start_link(Conf) when ?IS_BETA_NODE_CONF_VALID(Conf) ->
    gen_server:start_link(
	  {local, Conf#beta_node_conf.id}, 
	  ?MODULE,
	  [Conf], 
	  []);
start_link(BadConf) ->
	{error,{bad_conf,BadConf}}.

%% ====================================================================
%% Server functions
%% ====================================================================

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init([Conf]) ->
	Activator=beta_node_act_srv:server_name(Conf#beta_node_conf.id),
	case whereis(Activator) of
		undefined -> 
			{stop,{activator_process_not_alive,Activator}};
		ActivatorPid ->
			link(ActivatorPid),
			{ok, #state{conf=Conf,log_pipe=open_mdf_log(Conf)},2000}
	end.

%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call(Request, From, State) ->
	try call(Request, From, State) of
		R -> R
	catch
		throw:Reason -> {reply, {error,Reason}, State}
	end.

%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast(Msg, State) ->
	io:format("handle_cast(~p,~p)",[Msg, State]),
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info(timeout, State) ->
	LogPipeState=State#state.log_pipe,
	LogPipeState1=pipe:flush(LogPipeState),
	% activation here
	cast_activator(State),
	{noreply,
	 State#state{
				 log_pipe=LogPipeState1
				},
	 hibernate};
handle_info(_Info, State) ->
	{noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------
server_call(ServerRef,Request) ->
	case gen_server:call(ServerRef, Request, infinity) of
		{error,Reason} ->
			throw(Reason);
		Result ->
			Result
	end.

call(Request, From, State) 
  when is_record(Request, mdf_req)->
    Reply = ok,
    {reply, Reply, log_modification(Request, From, State),2000};
call(Request, From, State) ->
	throw({bad_call,{Request, From, State}}).

log_modification(Req,_From,State) ->
	LogPipeState=State#state.log_pipe,
	LogPipeState1=pipe:push(LogPipeState,Req),
	State#state{log_pipe=LogPipeState1}.

%% --------------------------------------------------------------------
%% Func: open_mdf_log(Conf) -> PipeState
%% Purpose: Opens a pipe to the beta node log. Note that this log 
%% 			wrapped by validating pipe that validates correctness 
%% 			of input data.
%% Returns: Pipe state.
%% --------------------------------------------------------------------
open_mdf_log(Conf) ->
	InpIds=Conf#beta_node_conf.in_ids,
	LogPipe=pipe_disk_logger:new([{name,Conf#beta_node_conf.id}]),
	ValidatorPipe=pipe_filter:new(
					fun(#mdf_req{mdf_code=Code,input_id=InpId,data=Data}) 
						 when Code==ins orelse Code==del->
							case lists:member(InpId, InpIds) of
								true ->
									case Data of
										{_K,_V}->
											true;
										[{_K,_V}|_T]->
											true;
										[]->
											false;
										_ -> 
											throw({unexp_data_format,Data})
									end;
								false ->
									throw({unexp_input_id,InpId,InpIds})
							end;
					   (BadReq) ->
							throw({bad_mdf,BadReq})
					end, LogPipe),
	pipe:open(ValidatorPipe).

ins_del(OpCode,InputId,BetaNodeId, KVList) when 
  (OpCode==ins orelse
  OpCode==del),
  is_atom(InputId),
  is_atom(BetaNodeId),
  is_list(KVList) ->
	InPipe=in_pipe(InputId, BetaNodeId),
	Iter=iter:from_list([{OpCode,K,V}||{K,V}<-KVList]),
	Pump=pump:new(Iter, InPipe),
	pump:init(Pump).

cast_activator(State) when is_record(State, state) ->
	Conf=State#state.conf,
	Log=BetaNodeId=Conf#beta_node_conf.id,
	File=renamed_log_file(Log,Conf),
	% 1. reopen log
	case disk_log:reopen(Log, File) of
		% 2a. reopen success
		ok ->
			% 2a.1. get activator server reference
			ActivatorSrv=beta_node_act_srv:server_name(BetaNodeId),
			% 2a.2. do cast for activation from log file
			gen_server:cast(ActivatorSrv, 
							{activate_from_log,
							 fun(ok)->
									 case file:delete(File) of
										 ok -> ok;
										 {error,Reason} ->
											 error_logger:error_report(
											   [{subject,{activated_log_cleanup_failed,File}},
												{reason,Reason}])
									 end;
								({error,_Reason})->
									 % Failover logic here
									 failover
							 end,
							 {filename,File}});
		% 2b. reopen failure handlig
		{error,Reason}->
			% 2b.1. just report for now
			error_logger:error_report(
			  [{subject,{reopen_log_failed,Log,File}},
			   {reason,Reason},
			   {state,State}])
	end.

renamed_log_file(Log,Conf) when 
  is_atom(Log),?IS_BETA_NODE_CONF(Conf) ->
	"./"++atom_to_list(Log)++"_r.LOG".

%% --------------------------------------------------------------------
%%% eUnit test functions
%% --------------------------------------------------------------------

input_test() ->
	Conf=#beta_node_conf
		 {
		  id='bn#1', 									% an atom identifying beta_node server
		  in_ids=['an#1','an#2','an#3'], 				% a list of atoms
		  out_pipe=pipe_console:new(), 					% an output pipe
		  beta_mem_fun=fun beta_memory_ets:open/1,		% beta memory factory fun -> #mh
		  red_fun=fun(_Modif,_K,_IterList)-> ok end,	% reduce function
		  red_mode = inc								% reduce mode (inc|aggr)
		 },
	start_link(Conf),
	InPipe=in_pipe('an#1', 'bn#1'),
	Iter=iter:from_list([{ins,K,"someValue"}||K<-lists:seq(1, 100)]),
	Pump=pump:new(Iter, InPipe),
	pump:init(Pump).